package com.burges.net.dataStream.waterMark

import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.windowing.time.Time

/**
  * 创建人    BurgessLee 
  * 创建时间   2020/1/27 
  * 描述     使用Timestamps Assigner中的Periodic Watermark模式
  */
object AscendingTimestampAssignerTimestampsAndWatermark {

	def main(args: Array[String]): Unit = {
		val environment = StreamExecutionEnvironment.getExecutionEnvironment

		//指定系统时间概念为EventTime
		environment.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
		val input = environment.fromCollection(List(("a", 1L, 1), ("b", 1L, 1), ("b", 3L, 1)))
		//使用系统默认的Ascending分配时间信息和Watermark
		val withTimestampsAndWatermarks = input.assignAscendingTimestamps( t => t._3)
		//对数据集进行窗口运算
		withTimestampsAndWatermarks.keyBy(0).timeWindow(Time.seconds(10)).sum("_2")

		environment.execute("streaming AscendingTimestampAssigner")

	}

}
